/*
 * Copyright 2010-2011 CGT&DC ISA RAS
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package websolve.workflow.runtime;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import websolve.workflow.runtime.util.MiscUtils;

public class Workflow extends Task {

    private List<Task> tasks;
    private InOutTask input;
    private InOutTask output;

    public Workflow(String signature) {
        super(signature);
        tasks = new CopyOnWriteArrayList<Task>();
        input = new InOutTask(getName() + "_input");
        output = new InOutTask(getName() + "_output");
        tasks.add(input);
        tasks.add(output);
        for (Entry<String, Parameter> paramEntry : inParams.entrySet()) {
            addInput(paramEntry.getKey(), paramEntry.getValue().getType());
        }
        for (Entry<String, Parameter> paramEntry : outParams.entrySet()) {
            addOutput(paramEntry.getKey(), paramEntry.getValue().getType());
        }
    }

    public Workflow addInput(String name, String type) {
        input.addParam(name, type);
        return this;
    }

    public Workflow addOutput(String name, String type) {
        output.addParam(name, type);
        return this;
    }

    public Workflow addInput(String name) {
        addInput(name, null);
        return this;
    }

    public Workflow addOutput(String name) {
        addOutput(name, null);
        return this;
    }

    public Parameter getInput(String name) {
        return input.getOutput(name);
    }

    public Object getInputValue(String name) {
        return input.getInput(name).getValue();
    }

    public Parameter getOutput(String name) {
        return output.getInput(name);
    }

    public Map<String, Parameter> getOutputs() {
        return output.getInputs();
    }

    public Object getOutputValue(String name) {
        return output.getOutput(name).getValue();
    }

    public Workflow setInput(String name, Parameter param) {
        input.setInput(name, param);
        return this;
    }

    public Workflow setInputValue(String name, Object value) {
        // setInput(name, new Parameter(null, value));
        input.setInputValue(name, value);
        return this;
    }

    public Workflow setOutput(String name, Parameter param) {
        output.setInput(name, param);
        return this;
    }

    public void addTask(Task task) {
        tasks.add(task);
    }

    public void removeTask(Task task) {
        tasks.remove(task);
    }

    public List<Task> getTasks() {
        List<Task> tasksOut = new ArrayList<Task>();
        for (Task task : tasks) {
            if (!(task instanceof InOutTask)) {
                tasksOut.add(task);
            }
        }
        return tasksOut;
    }

    @Override
    public void run() {
        setState(TaskState.RUNNING);
        ExecutorService exec = null;
        try {
            exec = Executors.newCachedThreadPool();
            boolean noProgress = false;
            while (!noProgress) {
                noProgress = true;
                for (Task task : tasks) {
                    if (task.getState().equals(TaskState.WAITING)) {
                        boolean isReady = true;
                        for (Parameter param : task.getInputs().values()) {
                            if (param.isConnected() && !param.isReady()) {
                                isReady = false;
                                break;
                            }
                        }
                        if (isReady) {
                            task.setState(TaskState.READY);
                            exec.execute(task);
                        } else {
                            boolean isCancelled = false;
                            boolean isFailed = false;
                            for (Parameter param : task.getInputs().values()) {
                                if (param.isCancelled()) {
                                    isCancelled = true;
                                    break;
                                }
                                if (param.isFailed()) {
                                    isFailed = true;
                                    break;
                                }
                            }
                            if (isCancelled) {
                                task.setState(TaskState.CANCELLED);
                            }
                            if (isFailed) {
                                task.setState(TaskState.FAILED);
                            }
                        }
                    }

                    // do we need to cancel a task?
                    if (!(task instanceof InOutTask)
                            && (task.getState().equals(TaskState.WAITING) || task.getState().equals(TaskState.READY) || task
                                    .getState().equals(TaskState.RUNNING))) {
                        boolean toCancel = true;
                        for (Parameter output : task.getOutputs().values()) {
                            if (output.getConsumerCount() > 0) {
                                toCancel = false;
                                break;
                            }
                        }
                        if (toCancel) {
                            task.setState(TaskState.CANCELLED);
                        }
                    }

                    if (!task.getState().equals(TaskState.DONE) && !task.getState().equals(TaskState.CANCELLED)
                            && !task.getState().equals(TaskState.FAILED)) {
                        noProgress = false;
                    }
                }
                Thread.sleep(300);
            }
            boolean isFailed = false;
            for (Task task : tasks) {
                if (task.getState().equals(TaskState.FAILED)) {
                    isFailed = true;
                    break;
                }
            }
            if (isFailed) {
                setState(TaskState.FAILED);
            } else {
                setState(TaskState.DONE);
            }
        } catch (Exception e) {
            e.printStackTrace();
            setError(MiscUtils.stackTraceToString(e));
            setState(TaskState.FAILED);
        } finally {
            if (exec != null)
                exec.shutdown();
        }
    }

    private class InOutTask extends Task {

        public InOutTask(String name) {
            super(name);
        }

        @Override
        public void run() {
            setState(TaskState.RUNNING);

            for (Entry<String, Parameter> outParamEntry : getOutputs().entrySet()) {
                outParamEntry.getValue().setValue(getInputValue(outParamEntry.getKey()));
            }

            setState(TaskState.DONE);
        }

        public void addParam(String name, String type) {
            super.addInput(name, type);
            super.addOutput(name, type);
        }
    }

}
